package com.linkstec.spark;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.AnalysisException;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import static org.apache.spark.sql.functions.col;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
 
public class demo4 {
	private static final Logger logger = LogManager.getLogger();
    private static String appName = "spark.sql.demo";
    private static String master = "local[*]";
 
    public static void main(String[] args) {
        //初始化SparkSession
        SparkSession spark = SparkSession
                .builder()
                .appName(appName)
                .master(master)
                .getOrCreate();
        //读取元数据文件
        Dataset<Row> df = spark.read().json("people.json");
        //自定义选择某些字段
//        df = df.select("name","age");
        //对年龄字段进行加1计算
//        df = df.select(col("name"), col("age").plus(1));
        // 筛选年龄大于19岁的记录
        df = df.filter(col("age").gt(19));
        //按照年龄计数
//        df = df.groupBy("age").count();
        //此外，我还可以使用原生sql来处理以上操作。首先我们要建立people视图
        /*df.createOrReplaceTempView("people");

		Dataset<Row> sqlDF = spark.sql("SELECT * FROM people");
		JavaRDD<Row> rdd = sqlDF.toJavaRDD();*/
        //全局临时视图与系统保存的数据库绑定global_temp，我们必须使用限定名称来引用它，例如SELECT * FROM global_temp.view1。
		try {
			df.createGlobalTempView("people");
		} catch (AnalysisException e) {
			e.printStackTrace();
		}
		Dataset<Row> sqlDF = spark.sql("SELECT * FROM global_temp.people");
		JavaRDD<Row> rdd = sqlDF.toJavaRDD();

		
        //生成rdd
//        JavaRDD<Row> rdd = df.toJavaRDD();
        //遍历
        rdd.foreach(new VoidFunction<Row>() {
			private static final long serialVersionUID = 1L;

			public void call(Row row) throws Exception {
				logger.info(row.toString());
            }
        });
 
        spark.stop();
    }
}
